package demo.delay;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.Method;
import java.time.Duration;
import java.util.Collections;
import java.util.concurrent.TimeUnit;


public class KafkaDelayConsumerThread extends Thread {

    private Logger log = LoggerFactory.getLogger(KafkaDelayConsumerThread.class);

    private KafkaConsumer<String, String> consumer;

    private Method method;

    private Object bean;

    private String topic;

    private long pauseTime;

    private long delayTime;

    private int retryTimes;

    private KafkaDltMessageHandler kafkaDltMessageHandler;

    public KafkaDelayConsumerThread(KafkaDelayConsumerContainer consumerContainer) {
        KafkaDelayListener kafkaDelayListener = consumerContainer.getKafkaDelayListener();

        kafkaDltMessageHandler = consumerContainer.getKafkaDltMessageHandler();

        this.consumer = consumerContainer.getConsumer();
        this.method = consumerContainer.getMethod();
        this.bean = consumerContainer.getBean();
        this.topic = kafkaDelayListener.topic();

        this.pauseTime = kafkaDelayListener.pauseTimeSec();
        this.delayTime = kafkaDelayListener.delayTimeSec();
        this.retryTimes = kafkaDelayListener.retryTimes();
    }


    @Override
    public void run() {
        // 订阅
        consumer.subscribe(Collections.singleton(topic));

        boolean paused = false;

        while (true) {
            if (paused) {
                try {
                    TimeUnit.MILLISECONDS.sleep(pauseTime);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                if (log.isDebugEnabled()) {
                    log.debug("KafkaDelayConsumerManager: topic: {}, 消费重启", topic);
                }

                consumer.resume(consumer.paused());
                paused = false;

            } else {

                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(200));
                for (ConsumerRecord<String, String> consumerRecord : records) {

                    TopicPartition topicPartition = new TopicPartition(consumerRecord.topic(), consumerRecord.partition());

                    if (consumerRecord.timestamp() > System.currentTimeMillis() - delayTime) {

                        if (log.isDebugEnabled()) {
                            log.debug("KafkaDelayConsumerManager: topic: {}, 消费暂停", topic);
                        }
                        paused = true;

                        consumer.pause(Collections.singleton(topicPartition));
                        consumer.seek(topicPartition, consumerRecord.offset());
                        break;
                    } else {

                        try {
                            retryConsume(consumerRecord, 0);
                            consumer.commitSync();
                        } catch (InterruptedException e) {
                            log.error("KafkaDelayConsumerManager: consume fail, topic: {}, message: {}, error: {}", topic, consumerRecord.value(), e.getMessage());
                        }
                    }
                }
            }
        }

    }


    private void retryConsume(ConsumerRecord<String, String> consumerRecord, int consumeTimes) throws InterruptedException {
        if (retryTimes > consumeTimes) {
            try {
                method.invoke(bean, consumerRecord);
            } catch (Exception e) {
                log.error("KafkaDelayConsumerManager: consume fail, topic: {}, message: {}, error: {}", topic, consumerRecord.value(), e);
                TimeUnit.MILLISECONDS.sleep(pauseTime);
                retryConsume(consumerRecord, ++consumeTimes);
            }
        } else {
            kafkaDltMessageHandler.putMessage(topic, consumerRecord.value());
        }
    }


}
